5.8 Long-term Memory
本节介绍 LangChain 中的长期记忆机制，实现跨会话的信息持久化。
什么是 Long-term Memory？
长期记忆（Long-term Memory） 基于 LangGraph 的持久化机制，实现跨会话的信息存储。
与短期记忆（会话内）不同，长期记忆在多次对话之间保持，用于存储：
- 用户偏好
- 历史洞察
- 学习到的信息
记忆存储架构
系统将记忆组织为 JSON 文档，采用层次结构：
┌─────────────────────────────────────────────────────────┐
│ Store │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Namespace: ("users", "user_123") │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Key: │ │ Key: │ │ Key: │ │ │
│ │ │ "prefs" │ │ "history"│ │ "notes" │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ {json} │ │ {json} │ │ {json} │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Namespace: ("org", "acme") │ │
│ │ ... │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘| 概念 | 说明 | 类比 |
|---|---|---|
| Namespace | 自定义组织文件夹 | 目录路径 |
| Key | 命名空间内的唯一标识符 | 文件名 |
| Value | JSON 文档 | 文件内容 |
Store 基本操作
创建 Store
python
from langgraph.store.memory import InMemoryStore
# 开发环境：内存存储
store = InMemoryStore()
# 带向量搜索的存储
from langchain_openai import OpenAIEmbeddings
store = InMemoryStore(
index={
"dims": 1536, # 向量维度
"embed": OpenAIEmbeddings(),
}
)基本操作
python
# 写入数据
store.put(
namespace=("users", "user_123"),
key="preferences",
value={"language": "zh", "theme": "dark"}
)
# 读取数据
item = store.get(("users", "user_123"), "preferences")
if item:
print(item.value) # {"language": "zh", "theme": "dark"}
# 搜索
results = store.search(
namespace=("users",),
filter={"language": "zh"}
)
# 带向量的语义搜索
results = store.search(
namespace=("knowledge",),
query="机器学习",
limit=5
)
# 删除
store.delete(("users", "user_123"), "preferences")在工具中读取长期记忆
python
from langchain_core.tools import tool
from langchain.agents import ToolRuntime
from dataclasses import dataclass
@dataclass
class Context:
user_id: str
@tool
def get_user_preferences(runtime: ToolRuntime[Context]) -> str:
"""获取用户偏好设置"""
user_id = runtime.context.user_id
if not runtime.store:
return "Store 不可用"
# 从长期记忆获取偏好
item = runtime.store.get(("users", user_id), "preferences")
if item:
prefs = item.value
return f"""
用户偏好:
- 语言: {prefs.get('language', '未设置')}
- 主题: {prefs.get('theme', '未设置')}
- 通知: {prefs.get('notifications', '未设置')}
"""
return "未找到用户偏好设置"
@tool
def get_user_history(runtime: ToolRuntime[Context]) -> str:
"""获取用户历史交互"""
user_id = runtime.context.user_id
if not runtime.store:
return "Store 不可用"
# 搜索用户的所有历史记录
results = runtime.store.search(
namespace=("history", user_id),
limit=10
)
if results:
history = []
for item in results:
history.append(f"- {item.value.get('action')}: {item.value.get('summary')}")
return "最近活动:\n" + "\n".join(history)
return "暂无历史记录"在工具中写入长期记忆
python
from langchain_core.tools import tool
from langchain.agents import ToolRuntime
from typing import TypedDict
from datetime import datetime
class UserPreferences(TypedDict):
language: str
theme: str
notifications: bool
@tool
def update_user_preferences(
language: str = None,
theme: str = None,
notifications: bool = None,
runtime: ToolRuntime[Context] = None
) -> str:
"""更新用户偏好设置"""
user_id = runtime.context.user_id
if not runtime.store:
return "Store 不可用"
# 获取现有偏好
existing = runtime.store.get(("users", user_id), "preferences")
prefs = existing.value if existing else {}
# 更新提供的字段
if language:
prefs["language"] = language
if theme:
prefs["theme"] = theme
if notifications is not None:
prefs["notifications"] = notifications
prefs["updated_at"] = datetime.now().isoformat()
# 保存到长期记忆
runtime.store.put(
("users", user_id),
"preferences",
prefs
)
return f"已更新偏好设置: {prefs}"
@tool
def remember_user_info(
key: str,
value: str,
runtime: ToolRuntime[Context]
) -> str:
"""记住用户提供的信息"""
user_id = runtime.context.user_id
if not runtime.store:
return "Store 不可用"
# 存储用户信息
runtime.store.put(
("user_info", user_id),
key,
{
"value": value,
"remembered_at": datetime.now().isoformat()
}
)
return f"已记住: {key} = {value}"
@tool
def recall_user_info(
key: str,
runtime: ToolRuntime[Context]
) -> str:
"""回忆用户信息"""
user_id = runtime.context.user_id
if not runtime.store:
return "Store 不可用"
item = runtime.store.get(("user_info", user_id), key)
if item:
return f"{key}: {item.value['value']}"
return f"我不记得 {key} 的信息"完整示例
python
from dataclasses import dataclass
from langchain.agents import create_agent, ToolRuntime
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver
from datetime import datetime
# 上下文定义
@dataclass
class UserContext:
user_id: str
user_name: str
# 创建带向量搜索的 Store
store = InMemoryStore(
index={
"dims": 1536,
"embed": OpenAIEmbeddings(),
}
)
# 工具定义
@tool
def save_note(
title: str,
content: str,
runtime: ToolRuntime[UserContext]
) -> str:
"""保存笔记到长期记忆"""
user_id = runtime.context.user_id
note_id = f"note_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
runtime.store.put(
("notes", user_id),
note_id,
{
"title": title,
"content": content,
"created_at": datetime.now().isoformat(),
}
)
return f"已保存笔记: {title}"
@tool
def search_notes(
query: str,
runtime: ToolRuntime[UserContext]
) -> str:
"""搜索笔记"""
user_id = runtime.context.user_id
# 语义搜索
results = runtime.store.search(
namespace=("notes", user_id),
query=query,
limit=5
)
if not results:
return "未找到相关笔记"
notes = []
for item in results:
note = item.value
notes.append(f"- {note['title']}: {note['content'][:100]}...")
return "找到的笔记:\n" + "\n".join(notes)
@tool
def get_all_notes(runtime: ToolRuntime[UserContext]) -> str:
"""获取所有笔记"""
user_id = runtime.context.user_id
results = runtime.store.search(
namespace=("notes", user_id),
limit=100
)
if not results:
return "暂无笔记"
notes = []
for item in results:
note = item.value
notes.append(f"- [{item.key}] {note['title']}")
return f"共 {len(notes)} 条笔记:\n" + "\n".join(notes)
@tool
def update_profile(
field: str,
value: str,
runtime: ToolRuntime[UserContext]
) -> str:
"""更新用户资料"""
user_id = runtime.context.user_id
# 获取现有资料
existing = runtime.store.get(("profiles",), user_id)
profile = existing.value if existing else {"name": runtime.context.user_name}
profile[field] = value
profile["updated_at"] = datetime.now().isoformat()
runtime.store.put(("profiles",), user_id, profile)
return f"已更新资料: {field} = {value}"
@tool
def get_profile(runtime: ToolRuntime[UserContext]) -> str:
"""获取用户资料"""
user_id = runtime.context.user_id
item = runtime.store.get(("profiles",), user_id)
if item:
profile = item.value
lines = [f"- {k}: {v}" for k, v in profile.items() if k != "updated_at"]
return "用户资料:\n" + "\n".join(lines)
return "暂无资料信息"
# 创建 Agent
agent = create_agent(
ChatOpenAI(model="gpt-4o"),
tools=[save_note, search_notes, get_all_notes, update_profile, get_profile],
context_schema=UserContext,
store=store,
checkpointer=InMemorySaver(),
system_prompt="""你是一个个人助手，可以帮助用户管理笔记和资料。
你可以:
- 保存和搜索笔记
- 更新和查看用户资料
记住用户告诉你的重要信息，以便将来使用。"""
)
# 使用
result = agent.invoke(
{"messages": [{"role": "user", "content": "保存一条笔记：明天下午3点开会"}]},
context=UserContext(user_id="user_001", user_name="张三"),
config={"configurable": {"thread_id": "session_1"}}
)
print(result["messages"][-1].content)
# 后续会话中可以搜索之前的笔记
result = agent.invoke(
{"messages": [{"role": "user", "content": "搜索关于开会的笔记"}]},
context=UserContext(user_id="user_001", user_name="张三"),
config={"configurable": {"thread_id": "session_2"}} # 新会话
)
print(result["messages"][-1].content)Store 类型对比
| Store | 持久化 | 向量搜索 | 适用场景 |
|---|---|---|---|
| InMemoryStore | ❌ | ✅ | 开发测试 |
| SqliteStore | ✅ | ❌ | 单机部署 |
| PostgresStore | ✅ | ✅ | 生产环境 |
短期 vs 长期记忆
| 方面 | 短期记忆 (State) | 长期记忆 (Store) |
|---|---|---|
| 作用域 | 单次会话 | 跨会话 |
| 存储 | Checkpointer | Store |
| 用途 | 对话历史、临时状态 | 用户偏好、学习信息 |
| 访问 | runtime.state | runtime.store |
最佳实践
| 实践 | 说明 |
|---|---|
| 合理的命名空间 | 使用 (类型, 用户ID) 结构 |
| 数据验证 | 写入前验证数据格式 |
| 定期清理 | 清理过期数据 |
| 隐私考虑 | 敏感信息加密存储 |
| 生产环境 | 使用 PostgresStore |
上一节：5.7 Retrieval